package com.neusoft.databus.core.hbase;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NavigableSet;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import javax.annotation.PostConstruct;

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageImpl;
import org.springframework.data.domain.Pageable;
import org.springframework.data.domain.Sort.Direction;
import org.springframework.data.domain.Sort.Order;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.stereotype.Service;

import com.google.common.collect.Maps;
import com.google.common.eventbus.AllowConcurrentEvents;
import com.google.common.eventbus.AsyncEventBus;
import com.google.common.eventbus.Subscribe;
import com.neusoft.bizcore.common.cache.CacheService;
import com.neusoft.bizcore.common.utils.CacheUtil;
import com.neusoft.bizcore.web.support.Searchable;
import com.neusoft.databus.common.bean.CommandBean;
import com.neusoft.databus.common.constant.DatabusConstant;
import com.neusoft.databus.common.constant.DatabusErrorCode;
import com.neusoft.databus.common.dto.MonitorDataDTO;
import com.neusoft.databus.common.utils.DefaultThreadFactory;
import com.neusoft.databus.core.metric.TpsEndpoint;
import com.neusoft.databus.core.service.MonitorDataService;

import lombok.extern.slf4j.Slf4j;

/**
 * 监控数据服务 <br>
 * 异步接收消息总线发送的消息 {#processMonitorDataEvent}
 *
 * @author sunchf
 */
@Slf4j
@Service
@ConditionalOnClass(HbaseTemplate.class)
@ConditionalOnProperty(name = "databus.core.hbase.enabled", havingValue = "true")
public class MonitorDataHBaseService implements MonitorDataService {

    @Autowired
    private HBaseProperties hbaseProperties;
    @Autowired
    private HbaseTemplate hbaseTemplate;
    @Autowired
    private MonitorDataHBaseConverter monitorDataHBaseConverter;

    @javax.annotation.Resource(name = "monitorDataEventBus")
    private AsyncEventBus monitorDataEventBus;
    @Autowired
    private CacheService cacheService;
    @Autowired
    private TpsEndpoint tpsEndpoint;

    @Value("${spring.application.name}")
    private String applicationName;
    @Value("${databus.core.batchSaveSize: 1000}")
    private int batchSaveSize;
    @Value("${databus.core.batchSaveMinInterval: 3000}")
    private int batchSaveMinInterval;

    private ArrayBlockingQueue<CommandBean> queue = null;
    private final ExecutorService executorService =
            Executors.newSingleThreadExecutor(new DefaultThreadFactory("hbase-queue"));

    @PostConstruct
    public void init() {
        this.monitorDataEventBus.register(this);

        this.queue = new ArrayBlockingQueue<>(50000);
        this.executorService.execute(new BatchSaveTask());
    }

    @Override
    public List<MonitorDataDTO> search(String resourceCode, Searchable searchable) {
        Long endTime = searchable.getLongValue("endTime");
        if (null == endTime) {
            endTime = new Date().getTime();
        }

        Long startTime = searchable.getLongValue("startTime");
        if ((null == startTime) || ((endTime - startTime) > 259200000)) {
            startTime = endTime - 259200000; // 3天
        }

        final String[] qualifiers = searchable.getStrArray("qualifier");

        final String startRow = resourceCode + startTime;
        final String stopRow = resourceCode + endTime;

        final Scan scan = new Scan();
        scan.setStartRow(Bytes.toBytes(startRow));
        scan.setStopRow(Bytes.toBytes(stopRow));

        if (null != qualifiers) {

            final Map<byte[], NavigableSet<byte[]>> familyMap =
                    new TreeMap<>(Bytes.BYTES_COMPARATOR);
            final NavigableSet<byte[]> set = new TreeSet<>(Bytes.BYTES_COMPARATOR);
            set.add(MonitorDataHBaseConverter.b_resource);
            set.add(MonitorDataHBaseConverter.b_collectedTime);
            Arrays.asList(qualifiers).stream().forEach(qualifer -> {
                set.add(Bytes.toBytes(qualifer));
            });
            familyMap.put(MonitorDataHBaseConverter.b_family, set);
            scan.setFamilyMap(familyMap);
        }

        final List<MonitorDataDTO> results = this.hbaseTemplate.execute(this.hbaseProperties.getTableName(), table -> {
            final List<MonitorDataDTO> dtos = new ArrayList<>();
            final ResultScanner rs = table.getScanner(scan);
            for (final Result result : rs) {
                final MonitorDataDTO dto = this.monitorDataHBaseConverter.toDTO(result);
                dtos.add(dto);
            }
            return dtos;
        });
        Collections.reverse(results);
        return results;
    }

    @Override
    public Page<MonitorDataDTO> search(final Pageable pageable, final Searchable searchable) {
        final String resourceCode = searchable.getStrValue("resourceCode");

        Long endTime = searchable.getLongValue("endTime");
        if (null == endTime) {
            endTime = new Date().getTime();
        }

        Long startTime = searchable.getLongValue("startTime");
        if ((null == startTime) || ((endTime - startTime) > 259200000)) {
            startTime = endTime - 259200000; // 3天
        }

        final String[] qualifiers = searchable.getStrArray("qualifier");

        final String startRow = resourceCode + startTime;
        final String stopRow = resourceCode + endTime;

        final Scan scan = new Scan();
        scan.setStartRow(Bytes.toBytes(startRow));
        scan.setStopRow(Bytes.toBytes(stopRow));

        if (null != qualifiers) {

            final Map<byte[], NavigableSet<byte[]>> familyMap =
                    new TreeMap<>(Bytes.BYTES_COMPARATOR);
            final NavigableSet<byte[]> set = new TreeSet<>(Bytes.BYTES_COMPARATOR);
            set.add(MonitorDataHBaseConverter.b_resource);
            set.add(MonitorDataHBaseConverter.b_collectedTime);
            Arrays.asList(qualifiers).stream().forEach(qualifer -> {
                set.add(Bytes.toBytes(qualifer));
            });
            familyMap.put(MonitorDataHBaseConverter.b_family, set);
            scan.setFamilyMap(familyMap);
        }

        final int pageSize = pageable != null ? (int) pageable.getPageSize() : 20;
        final int offset = pageable != null ? (int) pageable.getOffset() : 0;

        final List<Result> results = this.hbaseTemplate.execute(this.hbaseProperties.getTableName(), table -> {
            final ResultScanner rs = table.getScanner(scan);
            final Iterator<Result> it = rs.iterator();
            final List<Result> list = new ArrayList<>();
            while (it.hasNext()) {
                list.add(it.next());
            }
            return list;
        });

        final Order order = pageable.getSort().getOrderFor("collectedTime");
        Direction direction = Direction.DESC;
        if (null != order) {
            direction = order.getDirection();
        }

        if (Direction.DESC.equals(direction)) {
            Collections.reverse(results);
        }

        final List<Result> pageableResults = results.size() > 0
                ? results.subList(offset, (offset + pageSize) > results.size() ? results.size() : offset + pageSize)
                : new ArrayList<>();

        final List<MonitorDataDTO> pageableDTOs = pageableResults.stream().map(it -> {
            return this.monitorDataHBaseConverter.toDTO(it);
        }).collect(Collectors.toList());

        return new PageImpl<>(pageableDTOs, pageable, results.size());

    }

    /**
     * Event Bus事件处理
     *
     * @param commandBean
     * @return
     */
    @AllowConcurrentEvents // 线程安全
    @Subscribe
    public void processMonitorDataEvent(final CommandBean commandBean) {
        try {
            MDC.put("trace", commandBean.generateTraceId());
            MonitorDataHBaseService.log.info("monitor-data-eventbus: ready to process monitor data: {}",
                    commandBean.generateTraceId());
            if (commandBean.isError() && !DatabusErrorCode.ERR_CODE_406005.equals(commandBean.getErrorCode())) {
                MonitorDataHBaseService.log.info("monitor-data-eventbus: invalid monitor data, discard it.");
                return;
            }
            if (MonitorDataHBaseService.log.isDebugEnabled()) {
                MonitorDataHBaseService.log.debug("invoke tps endpoint metric collector, {}", this.tpsEndpoint);
            }
            this.tpsEndpoint.saveLatestTps(commandBean);
            this.saveCache(commandBean);

            if (!this.queue.offer(commandBean)) {
                MonitorDataHBaseService.log.warn("monitor-data-eventbus: queue is full. {} ",
                        commandBean.printBriefInfo());
            }
            MDC.clear();
        } catch (final Exception e) {
            MonitorDataHBaseService.log.error(e.getMessage(), e);
        }
    }

    private void saveBatch(final List<CommandBean> commandBeans) {

        try {
            final List<Put> puts = new ArrayList<>();
            commandBeans.stream().filter(it -> !it.isError()).forEach(c -> {
                puts.addAll(this.monitorDataHBaseConverter.toPut(c));
            });
            if (puts.size() == 0) {
                return;
            }

            final boolean result = this.hbaseTemplate.execute(this.hbaseProperties.getTableName(), table -> {
                try {
                    table.put(puts);
                } catch (final Throwable e) {
                    MonitorDataHBaseService.log.error(e.getMessage(), e);
                    return false;
                }
                return true;
            });

            if (result) {
                MonitorDataHBaseService.log.info("batch save {} data into hbase successfully", puts.size());
            } else {
                MonitorDataHBaseService.log.error("batch save {} data into hbase unsuccessfully", puts.size());
            }
        } catch (final Exception e) {
            MonitorDataHBaseService.log.error(e.getMessage(), e);
        }

    }

    private void saveCache(final CommandBean commandBean) {

        try {
            commandBean.getResourceBean().forEach(r -> {
                try {
                    final List<Map<String, Object>> metricData = r.getMetrics().stream().filter(i -> i.isParseSuccess())
                            .map(it -> {
                                final Map<String, Object> map = new HashMap<>();
                                map.put("name", it.getName());
                                map.put("value", it.getValue());
                                map.put("code", it.getCode());
                                if (null != it.getUnit()) {
                                    map.put("unit", it.getUnit());
                                }
                                return map;
                            }).collect(Collectors.toList());
                    final Map<String, Object> monitor = Maps.newHashMap();
                    monitor.put("collectedTime", commandBean.getCollectEndTime());
                    monitor.put("elapsedTime",
                            commandBean.getEndTime().getTime() - commandBean.getBeginTime().getTime());
                    monitor.put("data", metricData);
                    if (!metricData.isEmpty() && (metricData.size() > 0)) {
                        this.cacheService.put(DatabusConstant.CACHE_NAME_MONITOR_DATA,
                                this.getMonitorDataKey(r.getCode()),
                                monitor);
                    }
                } catch (final Exception e) {
                    MonitorDataHBaseService.log.error(e.getMessage(), e);
                }
            });
        } catch (final Exception e) {
            MonitorDataHBaseService.log.error(e.getMessage(), e);
        }

    }

    @Override
    @SuppressWarnings("unchecked")
    public Map<String, Object> findInCache(final String resourceCode) {
        Map<String, Object> data = Maps.newHashMap();
        final String key = this.getMonitorDataKey(resourceCode);
        if (this.cacheService.isInCache(DatabusConstant.CACHE_NAME_MONITOR_DATA, key)) {
            data = (Map<String, Object>) this.cacheService.get(DatabusConstant.CACHE_NAME_MONITOR_DATA, key);
        }
        data.put("code", resourceCode);
        return data;
    }

    private String getMonitorDataKey(final String resourceCode) {
        final StringBuffer key = new StringBuffer();
        key.append(CacheUtil.keyBuilder(resourceCode));
        return key.toString();
    }

    class BatchSaveTask implements Runnable {
        private long lastBatchSaveTime = -1;

        @Override
        public void run() {
            while (true) {
                final int size = MonitorDataHBaseService.this.queue.size();
                if (size == 0) {
                    try {
                        Thread.sleep(100);
                    } catch (final InterruptedException e) {
                        MonitorDataHBaseService.log.error("", e);
                    }
                    continue;
                }

                if (MonitorDataHBaseService.log.isDebugEnabled()) {
                    MonitorDataHBaseService.log.debug("{} in queues", size);
                }

                final boolean condition1 = size >= MonitorDataHBaseService.this.batchSaveSize;
                final boolean condition2 = (System.currentTimeMillis()
                        - this.lastBatchSaveTime) >= MonitorDataHBaseService.this.batchSaveMinInterval;

                if (MonitorDataHBaseService.log.isDebugEnabled()) {
                    MonitorDataHBaseService.log.debug("condition1:{}, condition2:{}, {}:{},{}:{}, ", condition1,
                            condition2, size, MonitorDataHBaseService.this.batchSaveSize,
                            System.currentTimeMillis() - this.lastBatchSaveTime,
                            MonitorDataHBaseService.this.batchSaveMinInterval);
                }

                final List<CommandBean> commandBeans = new ArrayList<>();
                if (condition1 || condition2) {
                    for (int i = 0; i < (MonitorDataHBaseService.this.batchSaveSize > size ? size
                            : MonitorDataHBaseService.this.batchSaveSize); i++) {
                        CommandBean bean = null;
                        try {
                            bean = MonitorDataHBaseService.this.queue.poll(3, TimeUnit.SECONDS);
                        } catch (final InterruptedException e) {
                            MonitorDataHBaseService.log.error(e.getMessage(), e);
                        }
                        if (null != bean) {
                            commandBeans.add(bean);
                        }
                    }
                }
                if (commandBeans.size() > 0) {
                    MonitorDataHBaseService.this.saveBatch(commandBeans);
                    this.lastBatchSaveTime = System.currentTimeMillis();
                }

            }
        }
    }

    @Override
    public Map<String, List<Map<String, Object>>> search(List<String> resourceCodes, long startTime, long endTime) {

        final Map<String, List<Map<String, Object>>> results = new HashMap<>();
        resourceCodes.forEach(code -> {
            final String startRow = code + startTime;
            final String stopRow = code + endTime;

            final Scan scan = new Scan();
            scan.setStartRow(Bytes.toBytes(startRow));
            scan.setStopRow(Bytes.toBytes(stopRow));

            final List<Map<String, Object>> monitorData =
                    this.hbaseTemplate.execute(this.hbaseProperties.getTableName(), table -> {
                        final ResultScanner rs = table.getScanner(scan);
                        final Iterator<Result> it = rs.iterator();
                        Result first = null, last = null;
                        while (it.hasNext()) {
                            final Result r = it.next();
                            if (null == first) {
                                first = r;
                            }
                            last = r;
                        }
                        if (null != first) {
                            final List<Map<String, Object>> data = Arrays.asList(first, last).stream().map(r -> {
                                return this.monitorDataHBaseConverter.toDTO(r);
                            }).map(monitor -> {
                                final Map<String, Object> map = new HashMap<>();
                                map.put("resourceCode", code);
                                map.put("collectedTime", monitor.getCollectedTime().getTime());
                                map.put("data", monitor.getData());
                                return map;
                            }).collect(Collectors.toList());

                            return data;

                        } else {
                            return null;
                        }

                    });

            if (null != monitorData) {
                results.put(code, monitorData);
            }
        });

        return results;
    }

    @Override
    public long countOfDay(String resourceCode, int day) {
        final Calendar calendar = Calendar.getInstance();
        calendar.set(Calendar.HOUR_OF_DAY, 0);
        calendar.set(Calendar.MINUTE, 0);
        calendar.set(Calendar.SECOND, 0);
        calendar.add(Calendar.DAY_OF_MONTH, day);

        final Date thatDay = calendar.getTime();
        calendar.add(Calendar.SECOND, 86399);
        final Date thatDayEnd = calendar.getTime();

        final Long endTime = thatDayEnd.getTime();
        final Long startTime = thatDay.getTime();

        final String startRow = resourceCode + startTime;
        final String stopRow = resourceCode + endTime;

        final Scan scan = new Scan();
        scan.setStartRow(Bytes.toBytes(startRow));
        scan.setStopRow(Bytes.toBytes(stopRow));

        final int count = this.hbaseTemplate.execute(this.hbaseProperties.getTableName(), table -> {
            final ResultScanner rs = table.getScanner(scan);
            final Iterator<Result> it = rs.iterator();
            final List<Result> list = new ArrayList<>();
            while (it.hasNext()) {
                list.add(it.next());
            }
            return list.size();
        });

        return count;
    }
}
